package com.bw.gmall.realjtime.dws.app;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realjtime.dws.function.AsyncDimFunction;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TradeProvinceOrderBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.DorisMapFunction;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.jar.JarEntry;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realjtime.dws.app
 * @className: DwsTradeProvinceOrderWindow
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/7 18:33
 * @version: 1.0
 */

public class DwsTradeProvinceOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeProvinceOrderWindow().start(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, Constant.DWS_TRADE_PROVINCE_ORDER_WINDOW, 4, 10020);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        SingleOutputStreamOperator<TradeProvinceOrderBean> getEtl = getEtl(dataStreamSource);
        SingleOutputStreamOperator<TradeProvinceOrderBean> getKeyById = getKeyById(getEtl);

        SingleOutputStreamOperator<TradeProvinceOrderBean> getReduce = getReduce(getKeyById);
        SingleOutputStreamOperator<TradeProvinceOrderBean> getProvinceName = getProvinceName(getReduce);
        getProvinceName.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DWS_TRADE_PROVINCE_ORDER_WINDOW));
    }

    private static SingleOutputStreamOperator<TradeProvinceOrderBean> getProvinceName(SingleOutputStreamOperator<TradeProvinceOrderBean> getReduce) {
        return AsyncDataStream.unorderedWait(getReduce, new AsyncDimFunction<TradeProvinceOrderBean>("dim_base_province") {
            @Override
            public void getResult(TradeProvinceOrderBean tradeProvinceOrderBean, JSONObject jsonObject) {
                tradeProvinceOrderBean.setProvinceName(jsonObject.getString("name"));
            }

            @Override
            public String getDimId(TradeProvinceOrderBean tradeProvinceOrderBean) {
                return tradeProvinceOrderBean.getProvinceId();
            }
        }, 5000, TimeUnit.MINUTES, 100);
    }

    private static SingleOutputStreamOperator<TradeProvinceOrderBean> getReduce(SingleOutputStreamOperator<TradeProvinceOrderBean> getKeyById) {
        return getKeyById.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeProvinceOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderBean>() {
                    @Override
                    public long extractTimestamp(TradeProvinceOrderBean tradeProvinceOrderBean, long l) {
                        return tradeProvinceOrderBean.getTs();
                    }
                }))
                .keyBy(new KeySelector<TradeProvinceOrderBean, String>() {
                    @Override
                    public String getKey(TradeProvinceOrderBean tradeProvinceOrderBean) throws Exception {
                        return tradeProvinceOrderBean.getProvinceId();
                    }
                }).window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5)))
                .reduce(new ReduceFunction<TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean reduce(TradeProvinceOrderBean t1, TradeProvinceOrderBean t2) throws Exception {
                        t1.setOrderAmount(t1.getOrderAmount().add(t2.getOrderAmount()));
                        t1.getOrderIdSet().addAll(t2.getOrderIdSet());
                        return t1;
                    }
                }, new ProcessWindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>.Context context, Iterable<TradeProvinceOrderBean> iterable, Collector<TradeProvinceOrderBean> collector) throws Exception {
                        TradeProvinceOrderBean bean = iterable.iterator().next();

                        bean.setStt(DateFormatUtil.tsToDateTime(context.window().getStart()));
                        bean.setEdt(DateFormatUtil.tsToDateTime(context.window().getEnd()));
                        bean.setCurDate(DateFormatUtil.tsToDate(System.currentTimeMillis()));

                        bean.setOrderCount((long) bean.getOrderIdSet().size());
                        collector.collect(bean);
                    }
                });
    }

    private static  SingleOutputStreamOperator<TradeProvinceOrderBean> getKeyById(SingleOutputStreamOperator<TradeProvinceOrderBean> getEtl) {
        return getEtl.keyBy(e -> e.getOrderDetailId())
                .process(new KeyedProcessFunction<String, TradeProvinceOrderBean, TradeProvinceOrderBean>() {
                    private ValueState<Boolean> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<Boolean> v = new ValueStateDescriptor<>("bo", Boolean.class);
                        v.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(50))
                                .useProcessingTime()
                                .updateTtlOnCreateAndWrite()
                                .neverReturnExpired()
                                .build());
                        valueState = getRuntimeContext().getState(v);
                    }

                    @Override
                    public void processElement(TradeProvinceOrderBean tradeProvinceOrderBean, KeyedProcessFunction<String, TradeProvinceOrderBean, TradeProvinceOrderBean>.Context context, Collector<TradeProvinceOrderBean> collector) throws Exception {
                        Boolean value = valueState.value();
                        if (value == null) {
                            valueState.update(true);
                            collector.collect(tradeProvinceOrderBean);
                        }
                    }
                });
    }

    private static SingleOutputStreamOperator<TradeProvinceOrderBean> getEtl(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.map(new MapFunction<String, TradeProvinceOrderBean>() {
            @Override
            public TradeProvinceOrderBean map(String s) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                HashSet<String> set = new HashSet<>();
                set.add(jsonObject.getString("order_id"));
                return TradeProvinceOrderBean.builder()
                        .orderDetailId(jsonObject.getString("id"))
                        .orderAmount(jsonObject.getBigDecimal("split_total_amount"))
                        .provinceId(jsonObject.getString("province_id"))
                        .ts(jsonObject.getLong("ts") * 1000)
                        .orderIdSet(set)
                        .build();
            }
        });
    }

}

